package net.qiqbframework.messaging.task;


import net.qiqbframework.common.Assert;
import net.qiqbframework.messaging.GenericResultMessage;
import net.qiqbframework.messaging.Message;
import net.qiqbframework.messaging.ResultMessage;

import javax.annotation.Nonnull;
import java.util.concurrent.Callable;
import java.util.function.Consumer;
import java.util.function.Function;

import static net.qiqbframework.messaging.GenericResultMessage.asResultMessage;


public class DefaultMessageTask<T extends Message<?>> extends AbstractMessageTask<T> {

    private final MessageProcessingContext<T> processingContext;

    public static <T extends Message<?>> DefaultMessageTask<T> startAndGet(T message) {
        DefaultMessageTask<T> uow = new DefaultMessageTask<>(message);
        uow.start();
        return uow;
    }

    public DefaultMessageTask(T message) {
        processingContext = new MessageProcessingContext<>(message);
    }

    @Override
    public <R> ResultMessage<R> executeWithResult(Callable<R> task, @Nonnull RollbackConfiguration rollbackConfiguration) {
        if (phase() == TaskPhase.NOT_STARTED) {
            start();
        }
        Assert.state(phase() == TaskPhase.STARTED,
                () -> String.format("The MessageWork has an incompatible phase: %s", phase()));
        R result;
        ResultMessage<R> resultMessage;
        try {
            plan();
        } catch (Exception e) {
            resultMessage = asResultMessage(e);

            return resultMessage;
        }
        try {
            result = task.call();
            if (result instanceof ResultMessage) {
                // noinspection Duplicates
                resultMessage = (ResultMessage<R>) result;
            } else if (result instanceof Message) {
                resultMessage = new GenericResultMessage<>(result, ((Message) result).getMetaData());
            } else {
                resultMessage = new GenericResultMessage<>(result);
            }
        } catch (Error | Exception e) {
            resultMessage = asResultMessage(e);
            if (rollbackConfiguration.rollBackOn(e)) {
                rollback(e);
                return resultMessage;
            }
        }
        setExecutionResult(new ExecutionResult(resultMessage));
        try {
            commit();
        } catch (Exception e) {
            resultMessage = asResultMessage(e);
        }
        return resultMessage;
    }

    @Override
    protected void setRollbackCause(Throwable cause) {
        setExecutionResult(new ExecutionResult(new GenericResultMessage<>(cause)));
    }

    @Override
    protected void notifyHandlers(TaskPhase phase) {
        processingContext.notifyHandlers(this, phase);
    }

    @Override
    protected void addHandler(TaskPhase phase, Consumer<MessageTask<T>> handler) {
        Assert.state(!phase.isBefore(phase()), () -> "Cannot register a listener for phase: " + phase
                + " because the Unit of Work is already in a later phase: " + phase());
        processingContext.addHandler(phase, handler);
    }

    @Override
    public T getMessage() {
        return processingContext.getMessage();
    }

    @Override
    public MessageTask<T> transformMessage(Function<T, ? extends Message<?>> transformOperator) {
        processingContext.transformMessage(transformOperator);
        return this;
    }

    @Override
    public ExecutionResult getExecutionResult() {
        return processingContext.getExecutionResult();
    }

    @Override
    protected void setExecutionResult(ExecutionResult executionResult) {
        processingContext.setExecutionResult(executionResult);
    }
}
